package io.adrainty.bolee.framework.mq.stream;

import com.rabbitmq.client.Channel;
import io.adrainty.bolee.framework.commons.exception.MessageAckException;
import io.adrainty.bolee.framework.mq.basic.MQMessage;
import io.adrainty.bolee.framework.mq.basic.MQMessageType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

import java.util.function.Consumer;

/**
 * <p>BaseStreamHandler</p>
 *
 * @author AdRainty
 * @version V1.0.0
 * @description BaseStreamHandler
 * @since 2025/7/30 12:34:22
 */

@Slf4j
public abstract class BaseStreamHandler implements Consumer<Message<MQMessage>> {

    @Override
    public void accept(Message<MQMessage> message) {
        MessageHeaders headers = message.getHeaders();
        Channel channel = headers.get("amqp_channel", Channel.class);
        Long deliveryTag = headers.get("amqp_deliveryTag", Long.class);

        MQMessage mqMessage = message.getPayload();
        Long messageId = mqMessage.getMessageId();
        log.info("[onMessage][线程编号:{} 消息ID：{}]", Thread.currentThread().getId(), messageId);
        doConsume(mqMessage);

        if (channel == null || deliveryTag == null) {
            log.warn("sendAckMsg error: channel == null || deliveryTag == null, is auto send ack?");
            return;
        }

        try {
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            throw new MessageAckException(e);
        }
    }

    /**
     * 消费消息
     *
     * @param message 消息
     */
    public abstract void doConsume(MQMessage message);

    /**
     * 接收消息类型
     *
     * @return 接收消息类型
     */
    public abstract MQMessageType acceptMsgType();

}
